package com.test.reactor.mq.rocket;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executor;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConfigurationProperties(prefix = "rocketmq.consumer")
@Slf4j
@Data
public class RocketMQConsumer {
	private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();

	private String namesrvAddr;
	private String groupName;
	private String topic;
	private String tag;
	private int consumeThreadMin;
	private int consumeThreadMax;

	@Autowired
	private Executor myServiceExecuto;

	@Bean
	@ConditionalOnProperty(name="rocketmq.enabled", matchIfMissing = true, havingValue="true")
	public DefaultMQPushConsumer getRocketMQConsumer() {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
		consumer.setNamesrvAddr(namesrvAddr);
		consumer.setConsumeThreadMin(consumeThreadMin);
		consumer.setConsumeThreadMax(consumeThreadMax);
		consumer.setVipChannelEnabled(false);
		// 我们自己实现的监听类
		MessageListen messageListen = new MessageListen("c1");
		consumer.registerMessageListener(messageListen);
		consumer.setMaxReconsumeTimes(1);
		try {
			consumer.subscribe(topic, tag);
			consumer.start();
			log.info("consume is start ,groupName:{},topic:{}", groupName, topic);
		} catch (MQClientException e) {
			log.error("consume start error");
			e.printStackTrace();
		}
		return consumer;
	}

	@Bean
	@ConditionalOnProperty(name="rocketmq.enabled", matchIfMissing = true, havingValue="true")
	public DefaultMQPullConsumer getRocketMQConsumer2() throws MQClientException {
		MessageListen messageListen = new MessageListen("c2");
		DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(groupName + "2");
		consumer.setNamesrvAddr(namesrvAddr);
		consumer.setMessageModel(MessageModel.CLUSTERING);
		consumer.setVipChannelEnabled(false);
		consumer.start();
		Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic);
		for (MessageQueue mq : mqs) {
			log.info("Consume from the queue:" + mq.toString());
			myServiceExecuto.execute(() -> {

				while (true) {
					try {
						PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
						log.info(pullResult.toString());
						switch (pullResult.getPullStatus()) {
						case FOUND:
							List<MessageExt> msgFoundList = pullResult.getMsgFoundList();
							messageListen.consumeMessage(msgFoundList, null);
							putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
							break;
						case NO_MATCHED_MSG:
							break;
						case NO_NEW_MSG:
							break;
						case OFFSET_ILLEGAL:
							break;
						default:
							break;
						}
					} catch (Exception e) {
						log.error("PullMessage", e);
					}
				}
			});
		}
		return consumer;
	}

	private static long getMessageQueueOffset(MessageQueue mq) {
		Long offset = OFFSE_TABLE.get(mq);
		if (offset != null)
			return offset;
		return 0;
	}

	private static void putMessageQueueOffset(MessageQueue mq, long offset) {
		OFFSE_TABLE.put(mq, offset);
	}
}
